package com.searchbyesandhbase.service.serviceImpl;

import com.searchbyesandhbase.entity.doc.ArticleDoc;
import com.searchbyesandhbase.mapper.ArticleMapper;
import com.searchbyesandhbase.service.ArticleService;

import com.searchbyesandhbase.utils.HBaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author LJK
 * @description
 * @date 2024/10/27
 */
@Service
@Slf4j
public class ArticleServiceImpl implements ArticleService {
    @Autowired
    private ArticleMapper articleMapper;
    @Autowired
    private HBaseUtil hBaseUtil;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private ElasticsearchOperations elasticsearchOperations;
    String tableName = "article";

    public void addArticleIndex(ArticleDoc article) {
        articleMapper.save(article);
    }

    /*
     * 将数据写入hbase，将rowKey写入redis，redis的数据后续会写入es中
     * hbase的put是天然幂等的，不需要额外处理
     * */
    @Override
    public void addArticleRowKeyAndCache(List<ArticleDoc> list) {
        String familyName = "info";

        list.forEach(articleDoc -> {
            String author = articleDoc.getAuthor();
            Integer id = articleDoc.getId();
            String title = articleDoc.getTitle();
            String describe = articleDoc.getDescribe();
            String content = articleDoc.getContent();
            List<Put> puts = new LinkedList<>();
            Put put = new Put(Bytes.toBytes(id.toString()));
            put.addColumn(Bytes.toBytes(familyName),
                    Bytes.toBytes("author"),
                    Bytes.toBytes(author));
            put.addColumn(Bytes.toBytes(familyName),
                    Bytes.toBytes("title"),
                    Bytes.toBytes(title));
            put.addColumn(Bytes.toBytes(familyName),
                    Bytes.toBytes("describe"),
                    Bytes.toBytes(describe));
            put.addColumn(Bytes.toBytes(familyName),
                    Bytes.toBytes("content"),
                    Bytes.toBytes(content));
            puts.add(put);
            try {
                //hBaseUtil.put2HBaseList(tableName, puts);
                redisTemplate.opsForList().rightPush("l_article_ids", id.toString());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }

    /*
     * 将redis中的rowKey取出来，用hbase读取获得具体数据写入es
     */
    public void createEsTableByRedis() {
        String rowKey = (String) redisTemplate.opsForList().leftPop("l_article_ids");
        try {
            while (rowKey != null) {
                Map<String, String> map = hBaseUtil.getFromHBase("article", rowKey);
                ArticleDoc articleDoc = new ArticleDoc();
                articleDoc.setId(Integer.parseInt(rowKey));
                articleDoc.setAuthor(map.get("author"));
                articleDoc.setTitle(map.get("title"));
                articleDoc.setDescribe(map.get("describe"));
                articleDoc.setContent(map.get("content"));
                articleMapper.save(articleDoc);
                rowKey = (String) redisTemplate.opsForList().leftPop("l_article_ids");
            }
        } catch (IOException e) {
            //保证数据不丢失
            redisTemplate.opsForList().rightPush("l_article_ids", rowKey);
            throw new RuntimeException(e);
        }
    }

    public void addArticlesIndex(List<ArticleDoc> list) {
        articleMapper.saveAll(list);
    }

    @Override
    public ArticleDoc getOneByIndex(Integer id) {
        return articleMapper.findById(id).orElse(new ArticleDoc());
    }

    @Override
    public Map<String, Object> searchByKeyword(String keyword, int page, int size) {
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        //根据关键字查询
        if (!StringUtils.isEmpty(keyword)) {
            queryBuilder = queryBuilder.withQuery(QueryBuilders.
                    multiMatchQuery(keyword, "title", "content", "describe"));
        }
        //高亮
        HighlightBuilder highlightBuilder = new HighlightBuilder().field("title").field("describe")
                .preTags("<span style='color:red'>").postTags("</span>");
        //分页+排序
        queryBuilder = queryBuilder.withPageable(PageRequest.of(page, size))
                .withSort(SortBuilders.scoreSort().order(SortOrder.DESC))
                .withHighlightBuilder(highlightBuilder);

        SearchHits<ArticleDoc> search = elasticsearchOperations.search(queryBuilder.build(), ArticleDoc.class);

        long totalHits = search.getTotalHits();

        HashMap<String, Object> map = new HashMap<>();
        map.put("count", totalHits);
        map.put("page", page);
        map.put("size", size);
        List<ArticleDoc> list = new ArrayList<>();
        search.getSearchHits().forEach(hit -> {

            ArticleDoc articleDoc = hit.getContent();
            // 获取高亮标题
            List<String> titleList = hit.getHighlightField("title");
            if (!titleList.isEmpty()) {
                StringBuilder title = new StringBuilder();
                for (String s : titleList) {
                    title.append(s);
                }
                articleDoc.setTitle(title.toString());
            }

            // 获取高亮描述
            List<String> describeList = hit.getHighlightField("describe");
            if (!describeList.isEmpty()) {
                StringBuilder describe = new StringBuilder();
                for (String s : describeList) {
                    describe.append(s);
                }
                articleDoc.setDescribe(describe.toString());
            }
            list.add(articleDoc);
        });
        map.put("data", list);
        return map;
    }

    @Override
    public Map<String, String> getOneByRowKey(String rowKey) throws IOException {
        return hBaseUtil.getFromHBase(tableName,rowKey);
    }
}
